Repository navigation
fix: keep event order, measure delays on the monotonic clock, cancel by marking - #665
JMerrill-Fairness wants to merge 8 commits into
Conversation
Both queues are PriorityQueues ordered by TriggerData's execution_time alone, so triggers with equal times came out in heap order rather than the order they were put. Two events stamped alike by a coarse clock could run out of order; twenty raised in one callback with one stamp ran as 0, 2, 6, 14, 19, 18, ... TriggerData now numbers each trigger as it is created and compares that number after execution_time, so triggers due together keep their order. The new tests stamp every trigger with one time and check the order on both engines and both queues. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Triggers were stamped with time.time() plus their delay and held until time.time() reached the stamp, so a step of the wall clock moved every due time: stepped back, it held even an undelayed event until the clock caught up; stepped forward, it ran delayed events early. Stamps and the due checks in both engines now use time.monotonic(). The new delayed-event test drives a fake monotonic clock in all three modules, so a stamp on one clock checked against another would hold the event and fail. The ordering tests now fix the monotonic clock instead of the wall clock. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
EventQueue.remove() filtered the PriorityQueue's underlying list. A filtered heap is not always a heap, so the triggers left could come out of due order: of ten delayed events due from 10 to 90 ms, cancelling the first ran the event due at 50 ms before the one due at 40 ms. The trigger's future was dropped with it, so on the async engine a coroutine awaiting the cancelled event's send() was never answered. cancel_event() now marks each trigger sent with the send_id as cancelled and answers its future with None. The triggers stay where they are, so the heap is never rearranged from outside, and is_empty() drops a cancelled trigger when it reaches the head, before the processing loop could check whether it is due. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Tests drawn from the change's own code rather than its description: - cancel_event() called from a callback, on both engines, and a caller that stops waiting, before or while it awaits, leaving its event cancellable; - a send_id matched exactly: empty, whitespace, line endings, NUL, case, a zero-width space, a byte-order mark, composed against decomposed accents, and ids up to 2048 characters, none of which cancels an event sent without one; - a cancel with nothing queued cancels nothing sent later, two ids cancelled in turn cancel each, and a cancelled trigger is marked rather than removed; - the order test cancels the first, a middle and the last event due, and the last one queued; - no trigger is stamped or checked on the wall clock, the engines read time.monotonic, and a delay in milliseconds comes due that many thousandths of a second later, for none, 0, 1, 50 and -50; - events due together keep their order for 0, 1, 2 and 20 of them. Each was checked against mutants of the code it covers: 31, every one failed by at least one test without waiting for a timeout. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
SonarCloud's S9083 asks for `@pytest.fixture` without parentheses, and this project's ruff configuration (fixture-parentheses = true) asks for them. The two clock fixtures are now `autouse=True`, which both accept: every test in each module runs on the substituted clock, so none reads the real one, and a test that moves the clock asks for it by name. The send_ids that held a zero-width space, a byte-order mark and accents as literal characters are escapes again, so the file is ASCII. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
A delayed event queued from a callback, while the processing loop runs, comes due on a clock that moves 10 ms with every reading, on both engines. A coroutine that stops awaiting a delayed event, before or while it awaits, does not stop the event, which fires in due order. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
No `type: ignore`, `noqa`, `pragma: no cover` or `pragma: no branch` is left in base.py, sync.py, async_.py or statemachine.py. - BaseEngine declared select_transitions, select_eventless_transitions, _select_transitions, _first_transition_that_matches, microstep and _exit_states as synchronous, and AsyncEngine overrode them with coroutines, which mypy rejected. They now belong to SyncEngine, and BaseEngine keeps what both engines share. - Each processing loop is split into its macrostep, the external events and one external event, and add_descendant_states_to_enter into its history, region and initial-state parts. - Removed as unreachable: BaseEngine.empty and pop, which nothing called; the _run_microstep wrappers, since microstep already sends every error through _handle_error; the second pass over the internal queue, since the macrostep empties it and spawning invocations puts events only on the external queue; and the checks for an exited state being None. - is_in_final_state takes the parallel case first, so an atomic state needs no branch of its own. - StateChart: a cast for the default Model, vars(self).update, an assert that narrows initial_state, and _repr_svg_ through the diagram formatter, which renders SVG the same way. New tests cover the two lines the pragmas hid: stop() still stops the engine when cancelling invocations fails, and a transition into a parallel state and a state inside one of its regions enters that state and not the region's initial one. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
SyncEngine.processing_loop took a caller_future it never used: only AsyncEngine.put attaches a future to an event, so the sync engine was always handed None. StateChart._processing_loop now passes the future to an AsyncEngine and calls the sync loop without one, and the sync loop no longer takes it (SonarCloud python:S1172). The new SCXML test marks its module with `pytestmark` rather than `@pytest.mark.scxml()`, which this project's ruff configuration (mark-parentheses = true) and SonarCloud's python:S9083 disagree on. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
|
There was a problem hiding this comment.
Hi @JMerrill-Fairness, welcome, and thank you for your first contribution to python-statemachine! This is an impressive PR, and I want to say why before getting to the details.
The three fixes are real runtime bugs that had been sitting in the engines since the statechart work landed: heap order leaking into event order, cancel_event() filtering the list behind the heap, and delays on the wall clock. You did not just fix them, you reproduced each one with a test that failed before its fix, picked the textbook solution for each (lazy deletion plus a sequence tie-breaker is exactly what the heapq docs recommend), and wrote release notes a user can understand.
The refactor is the part I appreciate most. BaseEngine was declaring sync methods that AsyncEngine overrode with coroutines, and four type: ignore[override] hid that. Now the base holds only what both engines share, each loop is split into named steps, and not one type: ignore, noqa or pragma is left in the four modules. I checked every "unreachable" claim against the code and each one holds. Mutation-testing your own tests and keeping the fixes in separate commits made the review a pleasure.
The inline comments are all non-blocking, so I'm approving as is; feel free to pick up any of them here or in a follow-up.
| def is_empty(self): | ||
| return self.queue.qsize() == 0 | ||
| """Whether the queue holds no trigger that has not been cancelled. Cancelled triggers at | ||
| the head are dropped first, so the :meth:`pop` that follows returns one that was not.""" | ||
| with self.queue.mutex: | ||
| heap = self.queue.queue | ||
| while heap and heap[0].cancelled: | ||
| heappop(heap) | ||
| return not heap |
There was a problem hiding this comment.
suggestion (non-blocking): Two cross-thread windows on the sync engine survive the marking: pop() does not check cancelled, so a cancel_event() landing between this is_empty() and the loop's pop() runs a trigger whose future was already answered with None; and a not-yet-due trigger is out of the heap between pop() and the re-put() in _take_external_events, so cancel() scans past it and the event fires later (reproduced by calling cancel_event from a patched put(_delayed=True)). A single pop_due(now) that, under queue.mutex, drops cancelled heads, peeks at heap[0].execution_time and pops only when due closes both windows and removes the pop/re-put churn per loop turn.
| def cancel(self, send_id: str): | ||
| """Cancel every trigger with this ``send_id``, and answer a caller awaiting one with | ||
| ``None``. Each stays where it is, so the heap keeps its order, until :meth:`is_empty` | ||
| drops it from the head.""" |
There was a problem hiding this comment.
praise: Marking instead of filtering keeps the heap untouched from outside PriorityQueue's own mutex, which is the lazy-deletion pattern the heapq docs recommend, with sequence as the tie-breaker they also prescribe. Worth keeping over a filter plus heapify.
| # Only the async engine gives an event a future, for its caller to await the result by. | ||
| if isinstance(self._engine, AsyncEngine): | ||
| result = self._engine.processing_loop(caller_future) | ||
| else: | ||
| result = self._engine.processing_loop() |
There was a problem hiding this comment.
suggestion (non-blocking): The facade now inspects the engine's concrete class to pick a call shape, so a test double or a future engine must subclass AsyncEngine to receive a future, and isawaitable(result) below already handles the two engines without knowing them. Keeping processing_loop(caller_future=None) on BaseEngine with SyncEngine ignoring it (its value was always None) restores one contract; a pragma-free way to satisfy S1172 is to name it _caller_future or document it as accepted for interface parity.
| def wall_clock(): | ||
| raise AssertionError("a trigger was stamped or checked on the wall clock") | ||
|
|
||
| monkeypatch.setattr("time.time", wall_clock) |
There was a problem hiding this comment.
suggestion (non-blocking): Patching the global time.time also hits logging.LogRecord.__init__, so under the documented debug run (uv run pytest -o log_cli_level=DEBUG tests/test_statechart_delayed.py) every _debug line of the engine raises this assertion and the test fails. The per-module patches on the next lines already cover every reader in statemachine/.
| monkeypatch.setattr("time.time", wall_clock) |
| @pytest.fixture(autouse=True) | ||
| def clock(monkeypatch): | ||
| """Run every test here on a monotonic clock that moves only when the test moves it.""" | ||
| now = [1000.0] | ||
| for module in _CLOCKS: | ||
| monkeypatch.setattr(f"{module}.monotonic", lambda: now[0]) | ||
| return now |
There was a problem hiding this comment.
suggestion (non-blocking): _CLOCKS and this patch loop are repeated in tests/test_statechart_delayed.py (lines 24, 127, 190 and 218). A module that starts reading monotonic has to be added to every copy, and a missed one stamps a trigger on the real clock and the test waits for the real delay. Moving this fixture to tests/conftest.py next to sm_runner serves both files.
…by marking (#665) Three fixes to how the engines order, time and cancel events, and a refactor that removes every suppression from the modules they touch. Events due at the same moment kept their order only by chance: both queues are PriorityQueues ordered by execution_time alone, so triggers stamped alike came out in heap order. TriggerData now numbers each trigger as it is created and compares that number after execution_time. Delays were measured on the wall clock, so a step of time.time() held or hastened every due time. Stamps and due checks now use time.monotonic(). Cancelling a delayed event filtered the list behind the heap, which is not always a heap afterwards, and dropped the trigger's future, so a coroutine awaiting it on the async engine was never answered. cancel_event() now marks each trigger sent with the send_id as cancelled and answers its future with None; the queue drops cancelled triggers from its head. EventQueue.pop_due() pops the head only once it is due, under the queue's mutex, so a trigger waiting on its delay never leaves the queue and a cancel from another thread always finds it. No type: ignore, noqa, pragma: no cover or pragma: no branch is left in engines/base.py, engines/sync.py, engines/async_.py or statemachine.py. The methods only the sync engine implements synchronously moved from BaseEngine to SyncEngine, each processing loop is split into its macrostep, the external events and one external event, and the code the pragmas hid is either removed as unreachable or covered by a test. Release notes in docs/releases/3.2.2.md. Co-authored-by: Fernando Macedo <fgmacedo@gmail.com> Signed-off-by: Fernando Macedo <fgmacedo@gmail.com>
|
Merged into GitHub does not let a maintainer push to a pull request from an organization-owned fork, so I squashed the branch locally and added three small commits on top rather than wait on a round trip:
Everything else is your work as submitted. This will ship in 3.2.2. |



Three fixes to how the engines order, time and cancel events, and a refactor that removes every suppression from the modules they touch. Each is its own commit, and they can be split into separate PRs if you prefer.
Events due at the same moment kept their order only by chance
Both queues are
PriorityQueues ordered byTriggerData.execution_timealone, so triggers with equal times come out in heap order, not the order they were put. A clock coarse enough to stamp two events alike runs them out of order. With every trigger stamped alike, twenty events raised in one callback ran as0, 2, 6, 14, 19, 18, ..., on both engines and both queues.TriggerDatanow numbers each trigger as it is created (itertools.count) and compares that number afterexecution_time, so triggers due together keep their order.Delays were measured on the wall clock
Triggers were stamped with
time.time()plus their delay and held untiltime.time()reached the stamp, so a step of the wall clock moved every due time. Stepped back by a time sync, it held even an undelayed event until the clock caught up. Stepped forward, it ran delayed events early. Stamps and the due checks in both engines now usetime.monotonic().Cancelling a delayed event could disorder the others and strand its caller
EventQueue.remove()filtered the list behind the heap, and a filtered heap is not always a heap. Of ten delayed events due from 10 to 90 ms, cancelling the first ran the event due at 50 ms before the one due at 40 ms. The trigger's future was dropped with it, so on the async engine a coroutine awaiting the cancelled event'ssend()was never answered. Both happen on 3.2.1.cancel_event()now marks each trigger sent with thesend_idas cancelled, in a newTriggerData.cancelledfield, and answers its future withNone. The triggers stay where they are, so nothing rearranges the heap from outsidePriorityQueue, andEventQueue.is_empty()drops cancelled triggers from the head, so the processing loop never checks whether one is due.EventQueue.remove()is renamedcancel().The suppressions in the touched modules
No
type: ignore,noqa,pragma: no coverorpragma: no branchis left inengines/base.py,engines/sync.py,engines/async_.pyorstatemachine.py. Each was replaced by fixing what it covered:BaseEnginedeclaredselect_transitions,select_eventless_transitions,_select_transitions,_first_transition_that_matches,microstepand_exit_statesas synchronous, andAsyncEngineoverrode them with coroutines, which is what the fourtype: ignore[override]hid. They now belong toSyncEngine, andBaseEnginekeeps what both engines share.add_descendant_states_to_enterinto its history, region and initial-state parts, which takes every function under the complexity limit.BaseEngine.empty()andpop(), which nothing called; the_run_microstep()wrappers, sincemicrostep()already sends every error through_handle_error(); the second pass over the internal queue, since the macrostep empties it and spawning invocations puts events only on the external queue; and the checks for an exited state beingNone.SyncEngine.processing_loop()no longer takes acaller_futureit never used. OnlyAsyncEngine.put()attaches a future to an event, soStateChart._processing_loop()now passes one to the async engine alone.is_in_final_state()takes the parallel case first, so an atomic state needs no branch of its own.StateChart: acastfor the defaultModel,vars(self).update(), anassertthat narrowsinitial_state(asfactory.pydoes), and_repr_svg_()through the diagram formatter, which renders SVG the same way.Two lines the pragmas hid now have tests:
stop()still stops the engine when cancelling invocations fails, and a transition into a parallel state and a state inside one of its regions enters that state and not the region's initial one.Tests
tests/test_event_ordering.pystamps every trigger with one time (by patchingstatemachine.event_data.monotonic) and checks the order on both engines and both queues, for 0, 1, 2 and 20 events. It failed before its fix.tests/test_cancel_event.pyruns every test on a clock that moves only when the test moves it.send_idcancels every event sent with it and nothing else. It is matched exactly: empty, whitespace, line endings, NUL, case, a zero-width space, a byte-order mark, composed against decomposed accents, and ids up to 2048 characters. It never cancels an event sent without one, and a cancel with nothing queued cancels nothing sent later.cancel_event()works from inside a callback, and a cancelled trigger is marked rather than removed.None, whether it is cancelled once or twice, and a caller that stops waiting, before or while it awaits, leaves its event cancellable.tests/test_statechart_delayed.pychecks that no trigger is stamped or checked on the wall clock (readingtime.timefails the test), that the engines readtime.monotonic, and that a delay in milliseconds comes due that many thousandths of a second later, for no delay, 0, 1, 50 and -50.tests/test_statechart_delayed.pyalso queues a delayed event from a callback, on a clock that moves 10 ms with every reading, and checks that a coroutine that stops awaiting a delayed event, before or while it awaits, does not stop it firing in due order.tests/test_invoke.py::TestEngineStopCleanupandtests/scxml/test_parallel_regions.pycover the two lines the pragmas hid.--cov-fail-under=100, along with ruff, mypy and pyright, through the pre-commit hooks.Release notes are in
docs/releases/3.2.2.md.🤖 Generated with Claude Code